#include "TcpClient.h"
#include <boost/bind.hpp>
#include <clusnet/base/WeakCallback.h>

using namespace clusnet::net;

TcpClient::TcpClient(boost::asio::io_context& io_context, const char* host, const char* port, const boost::asio::ip::tcp::resolver::protocol_type& proto, bool reconn, bool nodelay, bool keepAlive)
    : io_(io_context),
      strand_(io_context),
      conn_(nullptr),
      reconn_(reconn),
      reconnTimer_(io_context),
      nodelay_(nodelay),
      keepAlive_(keepAlive),
      host_(host),
      port_(port),
      proto_(proto),
      newConnCallback_(nullptr),
      closeCallback_(nullptr),
      writeCallback_(nullptr),
      messageCallback_(nullptr),
      state_(TcpClientState::UNCONNECTED)
{
}

TcpClient::~TcpClient()
{
    assert(state_ == TcpClientState::CLOSED && conn_ == nullptr);
}

void TcpClient::safeStartConnect()
{
    auto self = shared_from_this();
    boost::asio::dispatch(io_, boost::asio::bind_executor(strand_, [this, self] 
    {
        if (state_ == TcpClientState::CLOSED)
        {
            return;
        }

        assert(conn_ == nullptr && state_ == TcpClientState::UNCONNECTED);
        state_ = TcpClientState::CONNECTING;
        conn_ = std::make_shared<TcpConnection>(io_);
        boost::asio::async_connect(conn_->socket(), getEndPoints(), boost::asio::bind_executor(strand_, boost::bind(&TcpClient::handleConnect, self, boost::asio::placeholders::error, boost::asio::placeholders::endpoint)));
    }));
}

void TcpClient::handleConnect(const boost::system::error_code& ec, const boost::asio::ip::tcp::endpoint& endpoint)
{
    if (ec)
    {
        LOG(WARNING) << "connect failed. " << ec.message();
        safeReconnect();
        return;
    }

    if (state_ == TcpClientState::CLOSED)
    {
        safeReconnect();
        return;
    }
    
    // 防止循环引用
    auto self(shared_from_this());
    conn_->initConnect(nodelay_, keepAlive_, 
            clusnet::base::makeWeakCallback(self, &TcpClient::handleWriteComplete),
            clusnet::base::makeWeakCallback(self, &TcpClient::handleMessage),
            clusnet::base::makeWeakCallback(self, &TcpClient::handleClose));

    state_ = TcpClientState::CONNECTED;
    if (newConnCallback_)
    {
        newConnCallback_(conn_);
    }
    else
    {
        conn_->safeStartReceive();
    }
}

void TcpClient::handleClose(TcpConnectionPtr& conn)
{
    assert(conn == conn_);
    if (closeCallback_)
    {
        closeCallback_(conn);
    }
    safeReconnect();
}

void TcpClient::handleMessage(TcpConnectionPtr& conn, TcpUniqDataPacketPtr&& data)
{
    assert(messageCallback_ != nullptr);
    messageCallback_(conn, std::move(data));
}

void TcpClient::handleWriteComplete(TcpConnectionPtr& conn, const TcpDataPacketPtr& data)
{
    if (writeCallback_)
    {
        writeCallback_(conn, data);
    }
}

void TcpClient::safeReconnect()
{
    auto self(shared_from_this());
    boost::asio::dispatch(io_, boost::asio::bind_executor(strand_, [this, self] 
    {
        conn_.reset();
        if (state_ == TcpClientState::CLOSED)
        {
            return;
        }
        state_ = TcpClientState::UNCONNECTED;
        if (reconn_)
        {
            reconnTimer_.expires_after(std::chrono::seconds(3));
            reconnTimer_.async_wait(boost::asio::bind_executor(strand_, [this, self](const boost::system::error_code& ec)
            {
                if (!ec)
                {
                    safeStartConnect();
                }
            }));
        }
    }));
}

void TcpClient::safeAsyncWrite(const TcpDataPacketPtr& data)
{
    auto self(shared_from_this());
    boost::asio::dispatch(io_, boost::asio::bind_executor(strand_, [this, self, data]
    {
        if (state_ != TcpClientState::CONNECTED)
        {
            return;
        }
        assert(conn_ != nullptr);
        conn_->safeAsyncWrite(data);
    }));
}

void TcpClient::safeClose()
{
    auto self(shared_from_this());
    boost::asio::dispatch(io_, boost::asio::bind_executor(strand_, [this, self]
    {
        if (state_ == TcpClientState::CLOSED)
        {
            return;
        }
        reconn_ = false;
        reconnTimer_.cancel();
        if (state_ == TcpClientState::CONNECTED)
        {
            assert(conn_ != nullptr);
            conn_->safeShutDown();
        }
        state_ = TcpClientState::CLOSED;
    }));
}
